package com.gitee.xhs.customFunction;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import static com.gitee.xhs.OperatorConstant.INTERVAL_JOIN_OUT_PUT_TAG;
import static com.gitee.xhs.OperatorConstant.SEPARATION_TRANSACTION_MAP_STATE;


/**
 * @author BigHead
 * @date 2022/7/20 14:16
 * @description OperatorSeparationTransactionProcess
 */
public class OperatorSeparationTransactionProcess extends KeyedProcessFunction<Long, Transaction, Transaction> {

    private transient MapState<Integer, Transaction> transactionMapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        transactionMapState = getRuntimeContext().getMapState(
                SEPARATION_TRANSACTION_MAP_STATE
        );
        super.open(parameters);
    }

    @Override
    public void processElement(Transaction value, KeyedProcessFunction<Long, Transaction, Transaction>.Context ctx, Collector<Transaction> out) throws Exception {
        if (transactionMapState.contains(value.getDataNum())) {
            ctx.output(INTERVAL_JOIN_OUT_PUT_TAG, value);
            transactionMapState.remove(value.getDataNum());
        } else {
            transactionMapState.put(value.getDataNum(), value);
            out.collect(value);
        }
    }

}
